﻿using Common.Extension;
using Common.Json;
using Common.MQ;
using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;

namespace Common.RabbitMQ
{
    /// <summary>
    /// RabbitMQ消息生产者
    /// </summary>
    public class RabbitProducer : IMQProducer
    {
        private ConnectionFactory factory;
        private IConnection connection;
        private IModel channel;
        private readonly RabbitConfig config;
        private readonly QueueSpec producerQueue;
        /// <summary>
        /// RabbitMQ消息生产者
        /// </summary>
        /// <param name="config"></param>
        /// <param name="queue"></param>
        public RabbitProducer(RabbitConfig config, QueueSpec queue)
        {
            this.config = config;
            this.producerQueue = queue;
            this.Initialize();
        }
        private void Initialize()
        {
            this.factory = new ConnectionFactory();
            this.factory.HostName = this.config.Host;
            this.factory.Port = this.config.Port;
            this.factory.UserName = this.config.UserName;
            this.factory.Password = this.config.Password;
            this.factory.AutomaticRecoveryEnabled = true;
            this.factory.TopologyRecoveryEnabled = true;
            this.factory.NetworkRecoveryInterval = TimeSpan.FromSeconds(10);
            if (!string.IsNullOrEmpty(this.producerQueue.VirtualHost))
            {
                this.factory.VirtualHost = this.producerQueue.VirtualHost;
            }
            this.connection = this.factory.CreateConnection();
            this.channel = this.connection.CreateModel();
            Declare(this.producerQueue);
        }
        /// <summary>
        /// 声明队列
        /// </summary>
        /// <param name="dest"></param>
        public void Declare(IDestination dest)
        {
            QueueSpec queue = dest as QueueSpec;
            string DLX_EXCHANGE = $"{queue.Exchange}-DLX";//死信交换机
            string DLX_QUEUE = $"{queue.Queue}-DLX";//死信队列
            var normalProperties = new Dictionary<string, object> { { "x-dead-letter-exchange", DLX_EXCHANGE } };
            //声明死信交换机
            channel.ExchangeDeclare(DLX_EXCHANGE, queue.ExchangeType.GetStringValue(), true, false, null);
            //声明普通交换机，普通交换机指定了死信交换机，当消费者Reject消息时，消息会被转发到死信交换机
            channel.ExchangeDeclare(queue.Exchange, queue.ExchangeType.GetStringValue(), true, false, null);

            //声明普通队列，消息抵达交换机之后若没有匹配的队列，消息会被丢弃，所以必须有生产者声明普通队列（当然，消费者声明也是可以的，但我们需要保证的消息发送之前必须有队列接收消息）
            channel.QueueDeclare(queue.Queue, true, false, false, normalProperties);
            //声明死信队列，每个交换机必须至少要有一个队列，否则交换机找不到匹配的队列会丢弃消息
            channel.QueueDeclare(DLX_QUEUE, true, false, false, null);
            //绑定普通队列到普通交换机
            channel.QueueBind(queue.Queue, queue.Exchange, queue.RoutingKey, null);
            //绑定死信队列到死信交换机
            channel.QueueBind(DLX_QUEUE, DLX_EXCHANGE, queue.RoutingKey, null);
        }
        /// <summary>
        /// 发送数据
        /// </summary>
        /// <param name="message"></param>
        /// <returns></returns>
        public async Task<bool> ProduceAsync(IMessageModel message)
        {
            return await Task.Run(() =>
            {
                string json = message.GetMessageJson();

                PublicationAddress addr = GetAddress(message.GetDestination());
                var properties = this.channel.CreateBasicProperties();
                properties.DeliveryMode = 2;
                properties.Type = message.GetMessageType();
                var msgProperties = message.GetProperties();
                if (msgProperties != null)
                {
                    properties.Expiration = (msgProperties.TimeoutSeconds * 1000).ToString();
                }
                byte[] body = Encoding.UTF8.GetBytes(json);
                this.channel.BasicPublish(addr, properties, body);
                return true;
            });
        }
        private PublicationAddress GetAddress(string desctination)
        {
            if (!PublicationAddress.TryParse(desctination, out PublicationAddress addr))
            {
                return new PublicationAddress(this.producerQueue.ExchangeType.GetStringValue(), producerQueue.Exchange, producerQueue.RoutingKey);
            }
            return new PublicationAddress(addr.ExchangeType ?? this.producerQueue.ExchangeType.GetStringValue(), addr.ExchangeName ?? producerQueue.Exchange, addr.RoutingKey ?? producerQueue.RoutingKey);
        }
        /// <summary>
        /// 发送消息
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="data">消息数据</param>
        /// <param name="destination">消息目的地（伪URI，格式：交换机类型://交换机名称/路由KEY  示例：topic://wallet-ex/#）</param>
        /// <returns></returns>
        public async Task<bool> ProduceAsync<T>(T data, string destination)
        {
            var message = new MessageModel<T>(destination, data);
            return await ProduceAsync(message);
        }
        /// <inheritdoc/>
        public async Task<bool> ProduceAsync<T>(T data, string destination, MessageProperty property)
        {
            var message = new MessageModel<T>(destination, data, property);
            return await ProduceAsync(message);
        }
    }
}
